package com.myFwk.service.serviceImpl;

import com.myFwk.constant.Constants;
import com.myFwk.service.MsgReceiveService;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.impl.AMQImpl;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.stereotype.Service;

import java.io.IOException;

/**
 * @author GL
 * @version 1.0
 * @date 2021/10/21 22:10
 */
@Service
@Slf4j
public class MsgReceiveServiceImpl implements MsgReceiveService {

    @RabbitListener(queues ={Constants.QUEUE_FWK_0,Constants.QUEUE_FWK_2})
    public void receiveMsg_queue0(Message content, Channel channel) throws IOException {
        AMQP.Confirm.SelectOk selectOk = channel.confirmSelect();

        boolean ack = true;
        try {
            String msg = new String(content.getBody());
            log.info("QUEUE_FWK_0接收到的消息：{}", msg);
            if (msg.contains("deadletter")) {
                throw new RuntimeException("死信队列异常");
            }
        } catch (Exception e) {
            ack = false;
        }
        if (!ack) {
            // TODO 当消息消费异常的时候 由于该队列邦定了死信队列 失败的消息都会存储在该死信队列当中。
            //目前确认机制存在问题，消费端确认之后无效
            log.error("消息消费发生异常，error msg");
            channel.basicNack(content.getMessageProperties().getDeliveryTag(), false, false);
        } else {
            /**
             * false 是否开启多条确认机制
             *
             */
            channel.basicAck(content.getMessageProperties().getDeliveryTag(), false);
//            channel.basicPublish();
//            channel.basicConsume()
        }


    }

    @RabbitListener(queues = Constants.QUEUE_FWK_1)
    public void receiveMsg_queue1(Message content, Channel channel) {
        try {
            channel.basicAck(content.getMessageProperties().getDeliveryTag(), false);
            log.info("QUEUE_FWK_1接收到的消息" + content.getMessageProperties().getDeliveryTag() + "：{}", new String(content.getBody()));
            /**
             * 模拟消息处理的
             */
            log.info("开始处理消息：{}", String.valueOf(content.getMessageProperties().getDeliveryTag()));
            /*
             *  TODO 其实可以通过速率的方式来控制消息--具体解决方案
             * 当某一时刻当消息速率达到一个值的时候选择存储在Redis，否则就进行redis进行缓冲
             *
             * */
            Thread.sleep(5000);
            log.info("消息处理结束：{}", String.valueOf(content.getMessageProperties().getDeliveryTag()));
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    @RabbitListener(queues = Constants.DELAY_QUEUE_FWK_0)
    public void receiveMsg_delay1(Message message, Channel channel) throws IOException {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        log.info("收到死信队列信息:" + message.getMessageProperties().getDeliveryTag() + "：{}", new String(message.getBody()));
    }
}
